package pipeline

import (
	"fmt"
	"sync"
	"time"
)

// DefaultProgressMonitor 默认进度监控器实现
type DefaultProgressMonitor struct {
	progress    map[string]*TaskProgress
	subscribers map[string][]chan *TaskProgress
	mutex       sync.RWMutex
}

// NewProgressMonitor 创建进度监控器
func NewProgressMonitor() ProgressMonitor {
	return &DefaultProgressMonitor{
		progress:    make(map[string]*TaskProgress),
		subscribers: make(map[string][]chan *TaskProgress),
	}
}

// UpdateProgress 更新进度
func (pm *DefaultProgressMonitor) UpdateProgress(taskID string, progress *TaskProgress) {
	pm.mutex.Lock()
	defer pm.mutex.Unlock()

	// 更新进度
	pm.progress[taskID] = progress

	// 通知订阅者
	if subscribers, exists := pm.subscribers[taskID]; exists {
		for _, ch := range subscribers {
			select {
			case ch <- progress:
			default:
				// 如果通道已满，跳过这次更新
			}
		}
	}
}

// GetProgress 获取进度
func (pm *DefaultProgressMonitor) GetProgress(taskID string) (*TaskProgress, error) {
	pm.mutex.RLock()
	defer pm.mutex.RUnlock()

	progress, exists := pm.progress[taskID]
	if !exists {
		return nil, fmt.Errorf("progress not found for task: %s", taskID)
	}

	// 返回副本
	progressCopy := *progress
	return &progressCopy, nil
}

// GetAllProgress 获取所有进度
func (pm *DefaultProgressMonitor) GetAllProgress() map[string]*TaskProgress {
	pm.mutex.RLock()
	defer pm.mutex.RUnlock()

	result := make(map[string]*TaskProgress)
	for taskID, progress := range pm.progress {
		progressCopy := *progress
		result[taskID] = &progressCopy
	}

	return result
}

// Subscribe 订阅进度更新
func (pm *DefaultProgressMonitor) Subscribe(taskID string) <-chan *TaskProgress {
	pm.mutex.Lock()
	defer pm.mutex.Unlock()

	ch := make(chan *TaskProgress, 10) // 缓冲通道

	if pm.subscribers[taskID] == nil {
		pm.subscribers[taskID] = make([]chan *TaskProgress, 0)
	}
	pm.subscribers[taskID] = append(pm.subscribers[taskID], ch)

	// 如果已有进度，立即发送
	if progress, exists := pm.progress[taskID]; exists {
		select {
		case ch <- progress:
		default:
		}
	}

	return ch
}

// Unsubscribe 取消订阅
func (pm *DefaultProgressMonitor) Unsubscribe(taskID string) {
	pm.mutex.Lock()
	defer pm.mutex.Unlock()

	if subscribers, exists := pm.subscribers[taskID]; exists {
		// 关闭所有通道
		for _, ch := range subscribers {
			close(ch)
		}
		delete(pm.subscribers, taskID)
	}
}

// CleanupCompleted 清理已完成的任务进度
func (pm *DefaultProgressMonitor) CleanupCompleted(maxAge time.Duration) {
	pm.mutex.Lock()
	defer pm.mutex.Unlock()

	now := time.Now()
	for taskID, progress := range pm.progress {
		if progress.Status == StatusCompleted || progress.Status == StatusFailed || progress.Status == StatusCancelled {
			if progress.EndTime != nil && now.Sub(*progress.EndTime) > maxAge {
				delete(pm.progress, taskID)

				// 清理订阅者
				if subscribers, exists := pm.subscribers[taskID]; exists {
					for _, ch := range subscribers {
						close(ch)
					}
					delete(pm.subscribers, taskID)
				}
			}
		}
	}
}

// GetStatistics 获取统计信息
func (pm *DefaultProgressMonitor) GetStatistics() map[string]int {
	pm.mutex.RLock()
	defer pm.mutex.RUnlock()

	stats := map[string]int{
		"total":      0,
		"pending":    0,
		"processing": 0,
		"completed":  0,
		"failed":     0,
		"cancelled":  0,
	}

	for _, progress := range pm.progress {
		stats["total"]++
		switch progress.Status {
		case StatusPending:
			stats["pending"]++
		case StatusProcessing:
			stats["processing"]++
		case StatusCompleted:
			stats["completed"]++
		case StatusFailed:
			stats["failed"]++
		case StatusCancelled:
			stats["cancelled"]++
		}
	}

	return stats
}
